package com.heiye.count.biz.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.heiye.count.api.dto.FindQuizCountsByIdReqDTO;
import com.heiye.count.api.dto.FindQuizCountsByIdRspDTO;
import com.heiye.count.api.dto.FindQuizCountsByIdsReqDTO;
import com.heiye.count.api.dto.UpdateQuizCommentCountReqDTO;
import com.heiye.count.biz.constants.RedisKeyConstants;
import com.heiye.count.biz.domain.dataobject.QuizCountDO;
import com.heiye.count.biz.domain.mapper.QuizCountDOMapper;
import com.heiye.count.biz.service.QuizCountService;
import com.heiye.framework.common.response.Response;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: heiye
 * @date: 2025/07/14 上午10:26
 * @version: v1.0.0
 * @description: 问题计数业务
 */
@Slf4j
@Service
public class QuizCountServiceImpl implements QuizCountService {

    @Resource
    private QuizCountDOMapper quizCountDOMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    /**
     * 批量查询问题计数数据
     *
     * @param findQuizCountsByIdsReqDTO
     * @return
     */
    @Override
    public Response<List<FindQuizCountsByIdRspDTO>> findQuizzesCountData(FindQuizCountsByIdsReqDTO findQuizCountsByIdsReqDTO) {
        // 需要查询的问题 ID 集合
        List<Long> quizIds = findQuizCountsByIdsReqDTO.getQuizIds();

        // 1. 先查询 Redis 缓存
        // 构建 Redis Hash Key 集合
        List<String> hashKeys = quizIds.stream()
                .map(RedisKeyConstants::buildCountQuizKey)
                .toList();

        // 使用 Pipelined 通道，从 Redis 中批量查询笔记 Hash 计数
        List<Object> countHashes = getCountHashesByPipelineFromRedis(hashKeys);

        // 返参 DTO
        List<FindQuizCountsByIdRspDTO> findQuizCountsByIdRspDTOS = Lists.newArrayList();

        // 用于存储缓存中不存在，需要查数据库的笔记 ID
        List<Long> quizIdsNeedQuery = Lists.newArrayList();

        // 循环入参中需要查询的笔记 ID 集合， 构建对应 DTO, 并设置缓存中已存在的计数，以及过滤出需要查数据库的笔记 ID
        for (int i = 0; i < quizIds.size(); i++) {
            Long quizId = quizIds.get(i);

            List<Integer> currCountHash = (List<Integer>) countHashes.get(i);

            // 点赞数，粉丝数，浏览数，评论数
            Integer likeTotal = currCountHash.get(0);
            Integer fansTotal = currCountHash.get(1);
            Integer browseTotal = currCountHash.get(2);
            Integer commentTotal = currCountHash.get(3);

            // Hash 中存在任意一个 Field 为 null，都需要查询数据库
            if (Objects.isNull(likeTotal) || Objects.isNull(fansTotal) || Objects.isNull(browseTotal) || Objects.isNull(commentTotal)) {
                quizIdsNeedQuery.add(quizId);
            }

            // 构建 DTO
            FindQuizCountsByIdRspDTO findQuizCountsByIdRspDTO = FindQuizCountsByIdRspDTO.builder()
                    .quizId(quizId)
                    .likeTotal(Objects.nonNull(likeTotal) ? Long.valueOf(likeTotal) : null)
                    .browseTotal(Objects.nonNull(browseTotal) ? Long.valueOf(browseTotal) : null)
                    .fansTotal(Objects.nonNull(fansTotal) ? Long.valueOf(fansTotal) : null)
                    .commentTotal(Objects.nonNull(commentTotal) ? Long.valueOf(commentTotal) : null)
                    .build();

            findQuizCountsByIdRspDTOS.add(findQuizCountsByIdRspDTO);
        }

        // 所有 Hash 计数都存在于 Redis 中，直接返参
        if (CollUtil.isEmpty(quizIdsNeedQuery)) {
            return Response.success(findQuizCountsByIdRspDTOS);
        }

        // 2. 若缓存中无，则查询数据库
        // 从数据库中批量查询过滤出的 quizIdsNeedQuery 笔记 ID
        List<QuizCountDO> quizCountDOS = quizCountDOMapper.selectByQuizIds(quizIdsNeedQuery);

        // 若数据库查询的记录不为空
        if (CollUtil.isNotEmpty(quizCountDOS)) {
            // DO 集合转 Map, 方便后续查询对应的笔记 ID 的计数
            Map<Long, QuizCountDO> quizIdAndDOMap = quizCountDOS.stream()
                    .collect(Collectors.toMap(QuizCountDO::getQuizId, quizCountDO -> quizCountDO));

            // 将笔记 Hash 计数同步到 Redis 中
            syncQuizHash2Redis(findQuizCountsByIdRspDTOS, quizIdAndDOMap);

            // 针对 DTO 中为 null 的计数字段，循环设置从数据库中查询到的计数
            for (FindQuizCountsByIdRspDTO findQuizCountsByIdRspDTO : findQuizCountsByIdRspDTOS) {
                Long quizId = findQuizCountsByIdRspDTO.getQuizId();
                Long likeTotal = findQuizCountsByIdRspDTO.getLikeTotal();
                Long fansTotal = findQuizCountsByIdRspDTO.getFansTotal();
                Long browseTotal = findQuizCountsByIdRspDTO.getBrowseTotal();
                Long commentTotal = findQuizCountsByIdRspDTO.getCommentTotal();

                QuizCountDO quizCountDO = quizIdAndDOMap.get(quizId);

                if (Objects.isNull(likeTotal)) {
                    findQuizCountsByIdRspDTO.setLikeTotal(Objects.nonNull(quizCountDO) ? quizCountDO.getLikeTotal() : 0);
                }
                if (Objects.isNull(fansTotal)) {
                    findQuizCountsByIdRspDTO.setFansTotal(Objects.nonNull(quizCountDO) ? quizCountDO.getFansTotal() : 0);
                }
                if (Objects.isNull(browseTotal)) {
                    findQuizCountsByIdRspDTO.setBrowseTotal(Objects.nonNull(quizCountDO) ? quizCountDO.getBrowseTotal() : 0);
                }
                if (Objects.isNull(commentTotal)) {
                    findQuizCountsByIdRspDTO.setCommentTotal(Objects.nonNull(quizCountDO) ? quizCountDO.getCommentTotal() : 0);
                }
            }
        }

        return Response.success(findQuizCountsByIdRspDTOS);
    }

    /**
     * 查询问题计数数据
     *
     * @param findQuizCountsByIdReqDTO
     * @return
     */
    @Override
    public Response<FindQuizCountsByIdRspDTO> findQuizCountData(FindQuizCountsByIdReqDTO findQuizCountsByIdReqDTO) {
        // 问题 ID
        Long quizId = findQuizCountsByIdReqDTO.getQuizId();

        // 1. 先查询 Redis 缓存
        String quizCountRedisKey = RedisKeyConstants.buildCountQuizKey(quizId);
        List<Object> counts = redisTemplate.opsForHash()
                .multiGet(quizCountRedisKey, List.of(
                        RedisKeyConstants.FIELD_LIKE_TOTAL,
                        RedisKeyConstants.FIELD_FANS_TOTAL,
                        RedisKeyConstants.FIELD_BROWSE_TOTAL,
                        RedisKeyConstants.FIELD_COMMENT_TOTAL
                ));

        // 若 Hash 中计数不为空，优先以其为主(实时性更高)
        Object likeTotal = counts.get(0);
        Object fansTotal = counts.get(1);
        Object browseTotal = counts.get(2);
        Object commentTotal = counts.get(3);

        // 构建返参
        FindQuizCountsByIdRspDTO findQuizCountsByIdRspDTO = FindQuizCountsByIdRspDTO.builder()
                .quizId(quizId)
                .likeTotal(Objects.isNull(likeTotal) ? 0 : Long.parseLong(String.valueOf(likeTotal)))
                .fansTotal(Objects.isNull(fansTotal) ? 0 : Long.parseLong(String.valueOf(fansTotal)))
                .browseTotal(Objects.isNull(browseTotal) ? 0 : Long.parseLong(String.valueOf(browseTotal)))
                .commentTotal(Objects.isNull(commentTotal) ? 0 : Long.parseLong(String.valueOf(commentTotal)))
                .build();

        // 若 Redis 中有任何一个计数为空
        boolean isAnyNull = counts.stream().anyMatch(Objects::isNull);

        // 2. 缓存中无，则查询数据库
        if (isAnyNull) {
            // 从数据库查询该问题的计数
            QuizCountDO quizCountDO = quizCountDOMapper.selectByQuizId(quizId);

            // 判断 Redis 中对应计数，若为空，则使用 DO 中的计数
            if (Objects.nonNull(quizCountDO) && Objects.isNull(likeTotal)) {
                findQuizCountsByIdRspDTO.setLikeTotal(quizCountDO.getLikeTotal());
            }
            if (Objects.nonNull(quizCountDO) && Objects.isNull(fansTotal)) {
                findQuizCountsByIdRspDTO.setFansTotal(quizCountDO.getFansTotal());
            }
            if (Objects.nonNull(quizCountDO) && Objects.isNull(browseTotal)) {
                findQuizCountsByIdRspDTO.setBrowseTotal(quizCountDO.getBrowseTotal());
            }
            if (Objects.nonNull(quizCountDO) && Objects.isNull(commentTotal)) {
                findQuizCountsByIdRspDTO.setCommentTotal(quizCountDO.getCommentTotal());
            }

            // 异步同步到 Redis 缓存中, 以便下次查询能够命中缓存
            syncHashCount2Redis(quizCountRedisKey, quizCountDO,
                    likeTotal, fansTotal, browseTotal, commentTotal);
        }


        return Response.success(findQuizCountsByIdRspDTO);
    }

    /**
     * 修改问题评论计数数据
     *
     * @param updateQuizCommentCountReqDTO
     * @return
     */
    @Override
    public Response<?> updateQuizCommentCountData(UpdateQuizCommentCountReqDTO updateQuizCommentCountReqDTO) {
        // 问题 ID
        Long quizId = updateQuizCommentCountReqDTO.getQuizId();
        // 需要修改的评论计数
        int count = updateQuizCommentCountReqDTO.getCount();
        // 构建 Redis Key
        String quizCountHasKey = RedisKeyConstants.buildCountQuizKey(quizId);

        boolean hasKey = redisTemplate.hasKey(quizCountHasKey);
        // 若存在，则修改问题评论数
        if (hasKey) {
            redisTemplate.opsForHash().increment(quizCountHasKey, RedisKeyConstants.FIELD_COMMENT_TOTAL, count);
        }
        // 修改问题评论数
        quizCountDOMapper.insertOrUpdateCommentTotalByQuizId(count, quizId);

        return Response.success();
    }

    /**
     * 将该问题的 Hash 计数同步到 Redis 中
     *
     * @param quizCountHasKey
     * @param quizCountDO
     * @param likeTotal
     * @param fansTotal
     * @param browseTotal
     * @param commentTotal
     */
    private void syncHashCount2Redis(String quizCountHasKey, QuizCountDO quizCountDO,
                                     Object likeTotal, Object fansTotal, Object browseTotal, Object commentTotal) {
        if (Objects.nonNull(quizCountDO)) {
            threadPoolTaskExecutor.submit(() -> {
                // 存放计数
                Map<String, Object> quizCountMap = Maps.newHashMap();

                if (Objects.isNull(likeTotal)) {
                    quizCountMap.put(RedisKeyConstants.FIELD_LIKE_TOTAL, Objects.isNull(quizCountDO.getLikeTotal()) ? 0 : quizCountDO.getLikeTotal());
                }
                if (Objects.isNull(fansTotal)) {
                    quizCountMap.put(RedisKeyConstants.FIELD_FANS_TOTAL, Objects.isNull(quizCountDO.getFansTotal()) ? 0 : quizCountDO.getFansTotal());
                }
                if (Objects.isNull(browseTotal)) {
                    quizCountMap.put(RedisKeyConstants.FIELD_BROWSE_TOTAL, Objects.isNull(quizCountDO.getBrowseTotal()) ? 0 : quizCountDO.getBrowseTotal());
                }
                if (Objects.isNull(commentTotal)) {
                    quizCountMap.put(RedisKeyConstants.FIELD_COMMENT_TOTAL, Objects.isNull(quizCountDO.getCommentTotal()) ? 0 : quizCountDO.getCommentTotal());
                }

                redisTemplate.executePipelined(new SessionCallback<>() {
                    @Override
                    public Object execute(RedisOperations operations) {
                        // 批量添加 Hash 的计数 Field
                        operations.opsForHash().putAll(quizCountHasKey, quizCountMap);

                        // 设置随机过期时间 (2 小时以内)
                        long expireTime = 60 * 60 + RandomUtil.randomInt(60 * 60);
                        operations.expire(quizCountHasKey, expireTime, TimeUnit.SECONDS);
                        return null;
                    }
                });
            });
        }
    }

    /**
     * 从 Redis 中批量查询问题 Hash 计数
     *
     * @param hashKeys
     * @return
     */
    private List<Object> getCountHashesByPipelineFromRedis(List<String> hashKeys) {
        return redisTemplate.executePipelined(new SessionCallback<>() {
            @Override
            public Object execute(RedisOperations operations) {
                for (String hashKey : hashKeys) {
                    // 批量获取多个字段
                    operations.opsForHash().multiGet(hashKey, List.of(
                                    RedisKeyConstants.FIELD_LIKE_TOTAL,
                                    RedisKeyConstants.FIELD_FANS_TOTAL,
                                    RedisKeyConstants.FIELD_BROWSE_TOTAL,
                                    RedisKeyConstants.FIELD_COMMENT_TOTAL
                            )
                    );
                }
                return null;
            }
        });
    }

    /**
     * 将问题 Hash 计数同步到 Redis 中
     *
     * @param findQuizCountsByIdRspDTOS
     * @param quizIdAndDOMap
     */
    private void syncQuizHash2Redis(List<FindQuizCountsByIdRspDTO> findQuizCountsByIdRspDTOS, Map<Long, QuizCountDO> quizIdAndDOMap) {
        // 将问题计数同步到 Redis 中
        redisTemplate.executePipelined(new SessionCallback<>() {

            @Override
            public Object execute(RedisOperations operations) {
                // 循环已经构建好的返参 DTO 集合
                for (FindQuizCountsByIdRspDTO findQuizCountsByIdRspDTO : findQuizCountsByIdRspDTOS) {
                    Long likeTotal = findQuizCountsByIdRspDTO.getLikeTotal();
                    Long fansTotal = findQuizCountsByIdRspDTO.getFansTotal();
                    Long browseTotal = findQuizCountsByIdRspDTO.getBrowseTotal();
                    Long commentTotal = findQuizCountsByIdRspDTO.getCommentTotal();

                    // 若当前 DTO 的所有计数不为空， 则无线同步 Hash
                    if (Objects.nonNull(likeTotal) && Objects.nonNull(fansTotal) && Objects.nonNull(browseTotal) && Objects.nonNull(commentTotal)) {
                        continue;
                    }

                    // 否则，若有任意一个 Field 计数为空，则需要同步对应的 Field
                    Long quizId = findQuizCountsByIdRspDTO.getQuizId();
                    // 构建 Hash Key
                    String quizCountHasKey = RedisKeyConstants.buildCountQuizKey(quizId);

                    // 设置 Field 计数
                    Map<String, Long> countMap = Maps.newHashMap();
                    QuizCountDO quizCountDO = quizIdAndDOMap.get(quizId);

                    if (Objects.isNull(likeTotal)) {
                        countMap.put(RedisKeyConstants.FIELD_LIKE_TOTAL,
                                Objects.nonNull(quizCountDO) ? quizCountDO.getLikeTotal() : 0);
                    }
                    if (Objects.isNull(fansTotal)) {
                        countMap.put(RedisKeyConstants.FIELD_FANS_TOTAL,
                                Objects.nonNull(quizCountDO) ? quizCountDO.getFansTotal() : 0);
                    }
                    if (Objects.isNull(browseTotal)) {
                        countMap.put(RedisKeyConstants.FIELD_BROWSE_TOTAL,
                                Objects.nonNull(quizCountDO) ? quizCountDO.getBrowseTotal() : 0);
                    }
                    if (Objects.isNull(commentTotal)) {
                        countMap.put(RedisKeyConstants.FIELD_COMMENT_TOTAL,
                                Objects.nonNull(quizCountDO) ? quizCountDO.getCommentTotal() : 0);
                    }

                    // 批量添加 Hash 的计数 Field
                    operations.opsForHash().putAll(quizCountHasKey, countMap);

                    // 设置随机过期时间 (1 小时以内)
                    long expireTime = 60 * 30 + RandomUtil.randomInt(60 * 30);
                    operations.expire(quizCountHasKey, expireTime, TimeUnit.SECONDS);
                }
                return null;
            }
        });
    }
}
